package com.lagou.rpc.consumer;

import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.consumer.proxy.RpcClientProxy;
import com.lagou.rpc.consumer.serverinfo.ServerInfo;
import com.lagou.zkutils.GetCuratorClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 *  测试类
 */
public class ClientBootStrap {

    //  实现负载均衡，客户端
    //    开启定时任务，定时将耗时比较长的服务端的响应时间设置为0
    //    开启线程，选择客户端进行请求操作

    public static ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);

    // 创建map集合，用来缓存(维护)服务器节点和rpc客户端之间的关系
    public static Map<String, RpcClient> clientMap = new HashMap<>();

    //   维护服务器节点和连接信息的关系 例如：127.0.0.1：8899，lastTime：XXX(上次执行时的系统时间)，UsedTime：XXX(响应时间)
    public static Map<String, ServerInfo> serverMap = new HashMap<>();

    public static void main(String[] args) throws Exception {
//        IUserService userService = (IUserService) RpcClientProxy.createProxy(IUserService.class);
//        User user = userService.getById(2);
//        System.out.println(user);

        //监听zk上指定路径的节点变化
        // 1.先获取zookeeper实例
        CuratorFramework zkClient = GetCuratorClient.getZkClient();
        // 2.监控节点的变化 (对哪个zkClient监听，监听的节点路径，这里是父节点)
        watchNode(zkClient,GetCuratorClient.path);

        // 3.定时任务，参数一：要执行的定时任务； 参数二：系统运行后延迟5s执行； 参数三：该任务执行完毕后延迟5S再次执行； 参数四：时间单位
        // 消费者在启动时创建定时线程池，每隔5s自动上报，更新zookeeper临时节点的值
        executorService.scheduleWithFixedDelay(new ScheduleTask(zkClient),5,5, TimeUnit.SECONDS);
        // 4.选择一个客户端执行任务
        new Thread(new SimpleTask(zkClient)).start();
        // 作业一不让main方法结束，一直休眠
//        Thread.sleep(Integer.MAX_VALUE);
    }

    /**
     *  监控zookeeper节点的变化
     */
    private static void watchNode(CuratorFramework zkClient, String path) throws Exception {
        TreeCache treeCache = new TreeCache(zkClient, path);
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
                ChildData data = treeCacheEvent.getData();
                System.out.println("data.getPath==="+data.getPath());
                String str = data.getPath().substring(data.getPath().lastIndexOf("/")+1); //例如：127.0.0.1:8000(一个服务器节点)
                System.out.println("str ==="+str);

                TreeCacheEvent.Type type = treeCacheEvent.getType();
                if (type == TreeCacheEvent.Type.NODE_ADDED){// 如果监听的节点变化类型是添加
                    // 获取到的就是当服务器在zookeeper上创建节点时候写入的时间信息，格式： 系统执行时间：响应时间
                    String value = new String(data.getData());
                    System.out.println("value===="+value);
                    String[] split = str.split(":");
                    if (split.length != 2){
                        return;
                    }

                    if (!serverMap.containsKey(str)){ // 如果容器中没有该服务器节点信息
                        String[] valueSplit = value.split(":");
                        if (valueSplit.length != 2){
                            return;
                        }
                        RpcClient rpcClient = new RpcClient(split[0], Integer.parseInt(split[1]));
                        clientMap.put(str, rpcClient);

                        // 已经建立好了连接
                        IUserService iUserService = (IUserService) RpcClientProxy.createProxy(IUserService.class, rpcClient);
                        ServerInfo serverInfo = new ServerInfo();
                        // 设置上一次任务执行的系统时间
                        serverInfo.setLastTime(Long.parseLong(valueSplit[0]));
                        // 设置响应时间
                        serverInfo.setUsedTime(Long.parseLong(valueSplit[1]));
                        serverInfo.setiUserService(iUserService);
                        // 把连接信息放入容器中
                        serverMap.put(str,serverInfo);
                    }
                }else if (type == TreeCacheEvent.Type.NODE_REMOVED){ // 节点删除
                    System.out.println("监听到有节点删除："+str);
                    clientMap.remove(str);
                    serverMap.remove(str);
                }

                System.out.println("------现有连接列表-----");
                Iterator<Map.Entry<String, RpcClient>> iterator = clientMap.entrySet().iterator();
                while (iterator.hasNext()){
                    Map.Entry<String, RpcClient> next = iterator.next();
                    System.out.println(next.getKey()+":"+next.getValue());
                }
            }
        });


        treeCache.start();
    }
}
